package main

import (
	"context"
	"fmt"
	"strconv"

	"github.com/asim/go-micro/v3"
	"github.com/asim/go-micro/v3/registry"
	"github.com/go-micro/plugins/v3/registry/etcd"
	"smeb.xyz/pub/websocket-subandpub-cluster/internal/handlerwebsocket/protos"
)

type GrpcSrvServiceHandler struct {
}

// // UserSomeoneMessage implements protos.WsMessageServiceHandler.
// func (*GrpcSrvServiceHandler) UserSomeoneMessage(ctx context.Context, req *protos.WsUserSomeoneMessageRequest, resp *protos.WsUserSomeoneMessageResponse) error {
// 	value, ok := IdentityConnections[req.Member]
// 	if ok {
// 		i, e := ZinxServer.GetConnMgr().Get(value)
// 		if e == nil {
// 			bytes := append([]byte{0, 0, 0, DATA,
// 				byte(len(req.Data) >> 24),
// 				byte(len(req.Data) >> 16),
// 				byte(len(req.Data) >> 8),
// 				byte(len(req.Data)),
// 			}, req.Data...)
// 			i.SendToQueue(bytes)
// 		}
// 	}
// 	resp.IsSuccess = true
// 	return nil
// }

// // 普通消息发送
// func (s *MessageServiceHandler) SendMessage(ctx context.Context, req *protos.WsMessageRequest, resp *protos.WsMessageResponse) (err error) {

// 	users, b := ChannelUserSubs[req.Channel]
// 	if b {
// 		for _, user := range users {
// 			i, e := ZinxServer.GetConnMgr().Get(user)
// 			if e == nil {
// 				// T L V
// 				bytes := append([]byte{0, 0, 0, DATA,
// 					byte(len(req.Data) >> 24),
// 					byte(len(req.Data) >> 16),
// 					byte(len(req.Data) >> 8),
// 					byte(len(req.Data)),
// 				}, req.Data...)
// 				i.SendToQueue(bytes)
// 			}
// 		}
// 	} else {
// 		ZapLogger.Error("channel not found")
// 		return fmt.Errorf("channel not found")
// 	}
// 	return nil
// }

// // 频道被解散
// func (s *MessageServiceHandler) OnChannelDeleted(ctx context.Context, req *protos.WsOnChannelDeletedRequest, resp *protos.WsOnChannelDeletedResponse) (err error) {

// 	_, b := ChannelUserSubs[req.Channel]
// 	if b {
// 		Redis.SRem(ctx, req.Channel, ServerName).Result()
// 		delete(ChannelUserSubs, req.Channel)
// 		return nil
// 	} else {
// 		ZapLogger.Error("channel not found")
// 		return fmt.Errorf("channel not found")
// 	}
// }

// // 频道成员被移除
// func (s *MessageServiceHandler) DeleteChannelMember(ctx context.Context, req *protos.WsDeleteChannelMembersRequest, resp *protos.WsDeleteChannelMembersResponse) (err error) {

// 	var membersId []uint64 = make([]uint64, 0)
// 	for _, member := range req.Members {
// 		conn, b := IdentityConnections[member]
// 		if b {
// 			membersId = append(membersId, conn)
// 		}
// 	}

// 	// delete membersId from ChannelUserSubs
// 	Mutex.Lock()
// 	for _, member := range membersId {
// 		for i, v := range ChannelUserSubs[req.Channel] {
// 			if v == member {
// 				ChannelUserSubs[req.Channel] = append(ChannelUserSubs[req.Channel][:i], ChannelUserSubs[req.Channel][i+1:]...)
// 				break
// 			}
// 		}
// 	}
// 	if len(ChannelUserSubs[req.Channel]) == 0 {
// 		Redis.SRem(ctx, req.Channel, ServerName).Result()
// 		delete(ChannelUserSubs, req.Channel)
// 	}
// 	Mutex.Unlock()
// 	return nil
// }

// // 用户订阅多个频道
// func (s *MessageServiceHandler) UserSubscribleChannels(ctx context.Context, req *protos.WsUserSubscribleChannelsRequest, resp *protos.WsUserSubscribleChannelsResponse) (err error) {
// 	channels := strings.Split(req.Channels, ",")

// 	connId, b := IdentityConnections[req.Member]
// 	if !b {
// 		return fmt.Errorf("member not found")
// 	}

// 	for _, channel := range channels {
// 		_, b := ChannelUserSubs[channel]
// 		if !b {
// 			ChannelUserSubs[channel] = []uint64{connId}
// 			Redis.SAdd(ctx, channel, ServerName).Result()
// 		} else {
// 			ChannelUserSubs[channel] = append(ChannelUserSubs[channel], connId)
// 		}
// 	}

// 	originChannels, b := UserChannels[req.Member]
// 	if !b {
// 		UserChannels[req.Member] = channels
// 	} else {
// 		UserChannels[req.Member] = append(originChannels, channels...)
// 	}

// 	return nil
// }

// // 频道成员E2E消息
// func (s *MessageServiceHandler) UserSomebodyMessage(ctx context.Context, req *protos.WsUserSomebodyMessageRequest, resp *protos.WsUserSomebodyMessageResponse) (err error) {

// 	users, b := ChannelUserSubs[req.Channel]
// 	if b {
// 		for _, user := range users {
// 			i, e := ZinxServer.GetConnMgr().Get(user)

// 			identity, err := i.GetProperty("identity")
// 			if err != nil {
// 				ZapLogger.Error("identity not found")
// 				continue
// 			} else {

// 				if e == nil && contains(req.Members, identity.(uint64)) {
// 					// T L V
// 					bytes := append([]byte{0, 0, 0, DATA,
// 						byte(len(req.Data) >> 24),
// 						byte(len(req.Data) >> 16),
// 						byte(len(req.Data) >> 8),
// 						byte(len(req.Data)),
// 					}, req.Data...)
// 					i.SendToQueue(bytes)
// 					break
// 				} else {
// 					continue
// 				}
// 			}

// 		}
// 	} else {
// 		ZapLogger.Error("channel not found")
// 		return fmt.Errorf("channel not found")
// 	}
// 	return nil
// }

func contains(list []uint64, value uint64) bool {
	for _, item := range list {
		if item == value {
			return true
		}
	}
	return false
}

// // 删除频道内成员
// func (*MessageServiceHandler) DeleteChannelMembers(ctx context.Context, req *protos.WsDeleteChannelMembersRequest, resp *protos.WsDeleteChannelMembersResponse) error {

// 	var membersId []uint64 = make([]uint64, 0)

// 	for _, identity := range req.Members {
// 		conn, b := IdentityConnections[identity]
// 		if b {
// 			membersId = append(membersId, conn)
// 		}
// 	}

// 	// reomve membersId from ChannelUserSubs
// 	Mutex.Lock()
// 	for _, member := range membersId {
// 		for i, v := range ChannelUserSubs[req.Channel] {
// 			if v == member {
// 				ChannelUserSubs[req.Channel] = append(ChannelUserSubs[req.Channel][:i], ChannelUserSubs[req.Channel][i+1:]...)
// 				break
// 			}
// 		}
// 	}
// 	if len(ChannelUserSubs[req.Channel]) == 0 {
// 		Redis.SRem(ctx, req.Channel, ServerName).Result()
// 		delete(ChannelUserSubs, req.Channel)
// 	}
// 	Mutex.Unlock()
// 	return nil
// }

// // 多用户订阅同一个频道
// func (*MessageServiceHandler) MultiUserSubscribleChannel(ctx context.Context, req *protos.WsMultiUserSubscribleChannelRequest, resp *protos.WsMultiUserSubscribleChannelResponse) error {

// 	for _, identity := range req.Members {
// 		connId, e := IdentityConnections[identity]

// 		if e {
// 			_, b := ChannelUserSubs[req.Channel]
// 			if !b {
// 				ChannelUserSubs[req.Channel] = []uint64{connId}
// 				Redis.SAdd(ctx, req.Channel, ServerName).Result()
// 			} else {
// 				ChannelUserSubs[req.Channel] = append(ChannelUserSubs[req.Channel], connId)
// 			}
// 		}
// 	}
// 	return nil
// }

func (*GrpcSrvServiceHandler) GetServerInfo(ctx context.Context, req *protos.GetServerInfoRequest, resp *protos.GetServerInfoResponse) error {
	cpu, allow, TotalAlloc, Sys, HeapInuse, percent := GetSysInfo()

	for _, v := range percent {
		resp.CpupercentList = append(resp.CpupercentList, v)
	}
	resp.Cpu = cpu
	resp.Alloc = allow
	resp.Totalalloc = TotalAlloc
	resp.Sys = Sys
	resp.HeapInuse = HeapInuse

	return nil

}

// 被下线
func (s *GrpcSrvServiceHandler) SendKickedOutMsg(ctx context.Context, req *protos.KickedOutRequest, resp *protos.KickedOutResponse) (err error) {
	i, e := ZinxServer.GetConnMgr().Get2(req.ConnectId)
	if e == nil {
		i.SendBuffMsg(KICKED, []byte{})
		i.Stop()
		return nil
	}
	return e
}

func (s *GrpcSrvServiceHandler) CreateNewChannel(ctx context.Context, req *protos.CreateNewChannelRequest, resp *protos.CreateNewChannelResponse) (err error) {
	panic("unimplemented")
}

func (s *GrpcSrvServiceHandler) DeleteChannel(ctx context.Context, req *protos.DeleteChannelRequest, resp *protos.DeleteChannelResponse) (err error) {
	_, b := ChannelUserSubs[req.Channel]
	if b {
		Redis.SRem(ctx, req.Channel, ServerName).Result()
		delete(ChannelUserSubs, req.Channel)
		resp.Code = 200
		resp.Msg = "删除频道成功"
		return nil
	} else {
		ZapLogger.Error("channel not found")
		resp.Code = 500
		resp.Msg = "频道不存在"
		return err
	}
}

func (s *GrpcSrvServiceHandler) GetChannelConnCount(ctx context.Context, req *protos.GetChannelConnCountRequest, resp *protos.GetChannelConnCountResponse) (err error) {
	channel := req.Channel
	data, e := Redis.SCard(ctx, channel).Result()
	if e != nil {
		resp.Code = 200
		resp.Count = data
		resp.Msg = "获取频道人数成功"
	} else {
		resp.Code = 500
		resp.Msg = "获取频道人数失败"
	}
	return nil
}

func (s *GrpcSrvServiceHandler) DeleteChannelMembers(ctx context.Context, req *protos.DeleteChannelMembersRequest, resp *protos.DeleteChannelMembersResponse) (err error) {

	var membersId []uint64 = make([]uint64, 0)
	for _, member := range req.Members {
		conn, b := IdentityConnections[member]
		if b {
			membersId = append(membersId, conn)
		}
	}

	// delete membersId from ChannelUserSubs
	Mutex.Lock()
	for _, member := range membersId {
		for i, v := range ChannelUserSubs[req.Channel] {
			if v == member {
				ChannelUserSubs[req.Channel] = append(ChannelUserSubs[req.Channel][:i], ChannelUserSubs[req.Channel][i+1:]...)
				break
			}
		}
	}
	if len(ChannelUserSubs[req.Channel]) == 0 {
		Redis.SRem(ctx, req.Channel, ServerName).Result()
		delete(ChannelUserSubs, req.Channel)
	}
	Mutex.Unlock()
	resp.Code = 200
	resp.Msg = "删除频道成员成功"
	return nil
}

func (s *GrpcSrvServiceHandler) MultipleIdentitySubscribleChannel(ctx context.Context, req *protos.MultipleIdentitySubscribleChannelRequest, resp *protos.MultipleIdentitySubscribleChannelResponse) (err error) {

	for _, identity := range req.Members {
		connId, e := IdentityConnections[identity]

		if e {
			_, b := ChannelUserSubs[req.Channel]
			if !b {
				ChannelUserSubs[req.Channel] = []uint64{connId}
				Redis.SAdd(ctx, req.Channel, ServerName).Result()
			} else {
				ChannelUserSubs[req.Channel] = append(ChannelUserSubs[req.Channel], connId)
			}
		}
	}
	resp.Code = 200
	resp.Msg = "订阅频道成功"
	return nil
}

func (s *GrpcSrvServiceHandler) PublishToChannel(ctx context.Context, req *protos.PublishToChannelRequest, resp *protos.PublishToChannelResponse) (err error) {

	users, b := ChannelUserSubs[req.Channel]
	if b {
		for _, user := range users {
			i, e := ZinxServer.GetConnMgr().Get(user)
			if e == nil {
				// T L V
				bytes := append([]byte{0, 0, 0, DATA,
					byte(len(req.Data) >> 24),
					byte(len(req.Data) >> 16),
					byte(len(req.Data) >> 8),
					byte(len(req.Data)),
				}, req.Data...)
				i.SendToQueue(bytes)
			}
		}
		resp.Code = 200
		resp.Msg = "成功"
		return nil
	} else {
		ZapLogger.Error("channel not found")
		resp.Code = 500
		resp.Msg = "频道不存在"
		return fmt.Errorf("channel not found")
	}
}

func (s *GrpcSrvServiceHandler) PublishToSomeoneInChannel(ctx context.Context, req *protos.PublishToSomeoneInChannelRequest, resp *protos.PublishToSomeoneInChannelResponse) (err error) {
	v, ok := IdentityConnections[req.Receiver]
	if ok {
		i, e := ZinxServer.GetConnMgr().Get(v)
		if e == nil {
			// T L V
			bytes := append([]byte{0, 0, 0, DATA,
				byte(len(req.Data) >> 24),
				byte(len(req.Data) >> 16),
				byte(len(req.Data) >> 8),
				byte(len(req.Data)),
			}, req.Data...)
			i.SendToQueue(bytes)
			resp.Code = 200
			resp.Msg = "成功"
			return nil
		}
	}
	resp.Code = 500
	resp.Msg = "失败"
	return fmt.Errorf("channel not found")
}

// PublishToSomebodyInChannel implements protos.GrpcServiceHandler.
func (s *GrpcSrvServiceHandler) PublishToSomebodyInChannel(ctx context.Context, req *protos.PublishToSomebodyInChannelRequest, resp *protos.PublishToSomebodyInChannelResponse) error {
	users, b := ChannelUserSubs[req.Channel]
	if b {
		for _, user := range users {
			i, e := ZinxServer.GetConnMgr().Get(user)

			identity, err := i.GetProperty("identity")
			if err != nil {
				ZapLogger.Error("identity not found")
				continue
			} else {

				if e == nil && contains(req.Receivers, identity.(uint64)) {
					// T L V
					bytes := append([]byte{0, 0, 0, DATA,
						byte(len(req.Data) >> 24),
						byte(len(req.Data) >> 16),
						byte(len(req.Data) >> 8),
						byte(len(req.Data)),
					}, req.Data...)
					i.SendToQueue(bytes)
					break
				} else {
					continue
				}
			}

		}
	} else {
		ZapLogger.Error("channel not found")
		return fmt.Errorf("channel not found")
	}
	return nil
}

func InitGrpcServer() {

	Service := micro.NewService(
		micro.Name(ServerName),
		micro.Registry(
			etcd.NewRegistry(
				registry.Addrs(Configs.EtcdUrls),
			),
		),
		micro.Address(":"+strconv.Itoa(Configs.GrpcPort)),
		micro.BeforeStop(
			// 注销
			CleanRegisterInfoBeforStop,
		),
	)

	protos.RegisterGrpcServiceHandler(Service.Server(), new(GrpcSrvServiceHandler))

	Service.Init()

	Service.Run()
}

func CleanRegisterInfoBeforStop() error {
	Redis.SRem(context.Background(), "wsserver", ServerName).Result()
	Redis.Del(context.Background(), ServerName).Result()
	for channel := range ChannelUserSubs {
		Redis.SRem(context.Background(), channel, ServerName).Result()
	}
	close(Tasks)
	close(SomebodyTasks)
	close(SomeoneTasks)
	return nil
}
